package com.study.chapter05.operator.sink;

import com.study.entity.Event;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @Description:
 * @Author: LiuQun
 * @Date: 2022/7/30 17:34
 */
public class Sink2MysqlTest {

    public static void main(String[] args) throws Exception {
        //环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> stream = env.fromElements(
                new Event("Bob", "./home", 1000L),
                new Event("Livy", "./home", 2000L),
                new Event("Alice", "./cart", 3000L),
                new Event("Bob", "./cart", 2000L),
                new Event("Livy", "./prod?id=20", 4000L),
                new Event("Bob", "./prod?id=30", 5000L),
                new Event("Bob", "./home", 6000L),
                new Event("Alice", "./home/car", 7000L)
        );
        //1.要执行的sql
        String sql = "INSERT INTO clicks(user,url) values(?,?) on duplicate key update url=(url)";
        //2.对sql中的参数进行替换的操作，类似PreparedStatement的功能
        JdbcStatementBuilder<Event> jdbcStatementBuilder = new JdbcStatementBuilder<Event>() {
            @Override
            public void accept(PreparedStatement ps, Event event) throws SQLException {
                ps.setString(1, event.user);
                ps.setString(2, event.url);
            }
        };
        //3.配置数据库连接的信息
        JdbcConnectionOptions connectionConfig = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withUrl("jdbc:mysql://localhost:3306/test_power")
                .withDriverName("com.mysql.jdbc.Driver")
                .withUsername("root")
                .withPassword("LQ3217103676")
                .build();

        //4.创建SinkFunction
        SinkFunction<Event> sinkFunction = JdbcSink.sink(sql, jdbcStatementBuilder, connectionConfig);

        //5.将数据写入到mysql中
        stream.addSink(sinkFunction);


        env.execute();
    }
}
